package kafka

import (
	"fmt"
	"github.com/Shopify/sarama"
)

func Consume(addr, tp string) {
	consumer, err := sarama.NewConsumer([]string{addr}, nil)
	if err != nil {
		panic(err)
	}
	partition, err := consumer.ConsumePartition(tp, 0, sarama.OffsetNewest)
	if err != nil {
		panic(err)
	}
	for m := range partition.Messages() {
		fmt.Println(m.Offset, string(m.Value))
	}
}
